import org.apache.spark.sql.{DataFrame, SparkSession}
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods._
import com.hankcs.hanlp.HanLP
import com.hankcs.hanlp.seg.common.Term
import scala.collection.JavaConverters._

object TextAnalysis {
  // 数据库连接配置
  private val jdbcUrl = "jdbc:mysql://localhost:3306/db_teaching?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Hongkong&allowPublicKeyRetrieval=true&allowMultiQueries=true"
  private val dbUser = "root"
  private val dbPassword = "20040429lk"

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("Analysis")
      .master("local[*]")
      .getOrCreate()

    try {
      // 读取和处理数据
      val filePath = getClass.getResource("/articles.txt").getPath
      println(s"Reading file from: $filePath")

      val textRDD = spark.sparkContext.textFile(filePath)

      // 解析JSON并提取contents字段
      val contentsRDD = textRDD.flatMap { line =>
        implicit val formats = DefaultFormats
        try {
          val json = parse(line)
          val contents = (json \ "contents").extractOpt[List[String]].getOrElse(Nil)
          contents
        } catch {
          case e: Exception =>
            println(s"Error: $line")
            Nil
        }
      }

      // 分词处理
      val wordsRDD = contentsRDD.flatMap { content =>
        HanLP.segment(content)
          .asScala
          .map(_.word)
          .filter(word => word.nonEmpty && word.length > 1)
      }

      // 词频统计
      import spark.implicits._
      val wordCountsDF = wordsRDD
        .map(word => (word, 1))
        .reduceByKey(_ + _)
        .toDF("word", "num")
        .sort($"num".desc)

      // 显示结果
      println("Top 100 frequent words:")
      wordCountsDF.show(100, truncate = false)

      // 保存结果到数据库
      saveAnalysisResult(wordCountsDF, "ads_word_num", "word VARCHAR(255), num INT")

    } catch {
      case e: Exception =>
        println(s"error: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      spark.stop()
    }
  }

  def saveAnalysisResult(data: DataFrame, dbtable: String, field: String): Unit = {

    try {
      data.write
        .format("jdbc")
        .option("url", jdbcUrl)
        .option("driver", "com.mysql.cj.jdbc.Driver")
        .option("dbtable", dbtable)
        .option("user", dbUser)
        .option("password", dbPassword)
        .option("createTableColumnTypes", field)
        .mode("overwrite")
        .save()

      println(s"成功: $dbtable")

    } catch {
      case e: Exception =>
        println(s"错误信息: ${e.getMessage}")
        throw e
    }
  }
}